package com.hm.kafkatest2.controller;

import com.hm.kafkatest2.dto.KafkaListenerRequest;
import com.hm.kafkatest2.dto.KafkaMessageRequest;
import com.hm.kafkatest2.service.KafkaListenerService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.web.bind.annotation.*;

import java.util.HashMap;
import java.util.Map;

@Slf4j
@RestController
@RequestMapping("/api/kafka")
@RequiredArgsConstructor
public class KafkaController {

    private final ProducerFactory<String, String> producerFactory;
    private final KafkaListenerService kafkaListenerService;

    /**
     * 测试Kafka集群连接
     */
    @PostMapping("/test-connection")
    public ResponseEntity<String> testConnection(@RequestBody Map<String, String> request) {
        try {
            String bootstrapServers = request.get("bootstrapServers");
            if (bootstrapServers == null || bootstrapServers.trim().isEmpty()) {
                return ResponseEntity.badRequest().body("Kafka集群地址不能为空");
            }
            
            // 使用带超时的连接检测方法
            String connectionError = kafkaListenerService.testConnection(bootstrapServers);
            if (connectionError != null) {
                return ResponseEntity.ok(connectionError);
            }
            
            return ResponseEntity.ok("success");
        } catch (Exception e) {
            log.error("测试连接异常", e);
            return ResponseEntity.status(500).body("测试连接异常: " + e.getMessage());
        }
    }

    @PostMapping("/send")
    public ResponseEntity<String> sendMessage(@RequestBody KafkaMessageRequest request) {
        org.apache.kafka.clients.producer.Producer<String, String> producer = null;
        try {
            // 先检测Kafka集群连接状态
            Map<String, Object> testConfigs = new HashMap<>(producerFactory.getConfigurationProperties());
            testConfigs.put("bootstrap.servers", request.getBootstrapServers());
            testConfigs.put("enable.idempotence", false);
            testConfigs.put("connections.max.idle.ms", 5000);
            testConfigs.put("request.timeout.ms", 5000);
            testConfigs.put("delivery.timeout.ms", 10000);
            
            ProducerFactory<String, String> testProducerFactory = new org.springframework.kafka.core.DefaultKafkaProducerFactory<>(testConfigs);
            KafkaTemplate<String, String> testKafkaTemplate = new KafkaTemplate<>(testProducerFactory);
            
            try {
                producer = testKafkaTemplate.getProducerFactory().createProducer();
                producer.close();
            } catch (Exception connectionException) {
                log.error("无法连接到Kafka集群: {}", request.getBootstrapServers(), connectionException);
                // 清理资源
                testKafkaTemplate.destroy();
                return ResponseEntity.status(500).body("无法连接到Kafka集群: " + request.getBootstrapServers() + 
                        "，请检查集群地址和网络连接。错误信息: " + connectionException.getMessage());
            }
            // 清理测试资源
            testKafkaTemplate.destroy();
            
            // 创建特定Kafka集群的生产者配置
            Map<String, Object> producerConfigs = new HashMap<>(producerFactory.getConfigurationProperties());
            producerConfigs.put("bootstrap.servers", request.getBootstrapServers());
            
            // 禁用幂等性以解决旧版本Kafka broker兼容性问题
            producerConfigs.put("enable.idempotence", false);
            
            // 设置较短的连接超时时间，避免长时间等待
            producerConfigs.put("connections.max.idle.ms", 5000);
            producerConfigs.put("request.timeout.ms", 5000);
            producerConfigs.put("delivery.timeout.ms", 10000);
            
            // 如果提供了groupId，则设置消费者组ID
            if (request.getGroupId() != null && !request.getGroupId().isEmpty()) {
                producerConfigs.put("group.id", request.getGroupId());
            }
            
            // 创建专用的KafkaTemplate
            ProducerFactory<String, String> customProducerFactory = new org.springframework.kafka.core.DefaultKafkaProducerFactory<>(producerConfigs);
            KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(customProducerFactory);
            
            // 发送消息到指定topic
            kafkaTemplate.send(request.getTopic(), request.getMessage());
            
            // 清理资源
            kafkaTemplate.destroy();
            
            log.info("消息已发送到Kafka集群: {}, topic: {}, group: {}", 
                    request.getBootstrapServers(), request.getTopic(), request.getGroupId());
            
            return ResponseEntity.ok("消息已成功发送到Kafka");
        } catch (Exception e) {
            log.error("发送消息到Kafka失败", e);
            return ResponseEntity.status(500).body("发送消息失败: " + e.getMessage());
        }
    }
    
    /**
     * 开启对指定Kafka集群和topic的监听
     */
    @PostMapping("/listen/start")
    public ResponseEntity<String> startListening(@RequestBody KafkaListenerRequest request) {
        try {
            // 先检测Kafka集群连接状态
            String connectionError = kafkaListenerService.testConnection(request.getBootstrapServers());
            if (connectionError != null) {
                return ResponseEntity.status(500).body(connectionError);
            }
            
            boolean success = kafkaListenerService.startListening(
                    request.getBootstrapServers(), request.getTopic(), request.getGroupId());
            if (success) {
                return ResponseEntity.ok("成功启动监听: " + request.getBootstrapServers() + 
                        ", Topic: " + request.getTopic() + 
                        ", Group: " + (request.getGroupId() != null ? request.getGroupId() : "default"));
            } else {
                return ResponseEntity.status(500).body("启动监听失败: " + request.getBootstrapServers() + 
                        ", Topic: " + request.getTopic() + 
                        ", Group: " + (request.getGroupId() != null ? request.getGroupId() : "default"));
            }
        } catch (Exception e) {
            log.error("启动监听异常", e);
            return ResponseEntity.status(500).body("启动监听异常: " + e.getMessage());
        }
    }
    
    /**
     * 停止对指定Kafka集群和topic的监听
     */
    @PostMapping("/listen/stop")
    public ResponseEntity<String> stopListening(@RequestBody KafkaListenerRequest request) {
        try {
            String uniqueId = request.getUniqueId();
            if (uniqueId != null && !uniqueId.isEmpty()) {
                // 如果提供了唯一标识，则停止特定的监听器
                boolean success = kafkaListenerService.stopListeningById(
                        request.getBootstrapServers(), request.getTopic(), request.getGroupId(), uniqueId);
                if (success) {
                    return ResponseEntity.ok("成功停止指定监听: " + request.getBootstrapServers() + 
                            ", Topic: " + request.getTopic() + 
                            ", Group: " + (request.getGroupId() != null ? request.getGroupId() : "default") +
                            ", UniqueId: " + uniqueId);
                } else {
                    return ResponseEntity.status(500).body("停止指定监听失败: " + request.getBootstrapServers() + 
                            ", Topic: " + request.getTopic() + 
                            ", Group: " + (request.getGroupId() != null ? request.getGroupId() : "default") +
                            ", UniqueId: " + uniqueId);
                }
            } else {
                // 如果没有提供唯一标识，则停止所有相同参数的监听器
                int stoppedCount = kafkaListenerService.stopAllListening(
                        request.getBootstrapServers(), request.getTopic(), request.getGroupId());
                if (stoppedCount > 0) {
                    return ResponseEntity.ok("成功停止 " + stoppedCount + " 个监听: " + request.getBootstrapServers() + 
                            ", Topic: " + request.getTopic() + 
                            ", Group: " + (request.getGroupId() != null ? request.getGroupId() : "default"));
                } else {
                    return ResponseEntity.status(500).body("未找到匹配的监听器: " + request.getBootstrapServers() + 
                            ", Topic: " + request.getTopic() + 
                            ", Group: " + (request.getGroupId() != null ? request.getGroupId() : "default"));
                }
            }
        } catch (Exception e) {
            log.error("停止监听异常", e);
            return ResponseEntity.status(500).body("停止监听异常: " + e.getMessage());
        }
    }
    
    /**
     * 获取监听状态
     * 返回所有监听器的状态，包括具有相同参数的多个监听器实例
     */
    @GetMapping("/listen/status")
    public ResponseEntity<Map<String, Boolean>> getListeningStatus() {
        try {
            Map<String, Boolean> status = kafkaListenerService.getListeningStatus();
            return ResponseEntity.ok(status);
        } catch (Exception e) {
            log.error("获取监听状态异常", e);
            return ResponseEntity.status(500).build();
        }
    }
}